package com.need2

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.sql.types._

import scala.collection.mutable.ListBuffer

/**
  * Created by zhuang on 2018/3/4.
  */
object DealData_3 extends App {

  private val load: Config = ConfigFactory.load()

  val conf = new SparkConf().setMaster("local[*]").setAppName(this.getClass.getSimpleName)
    .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  val sc = new SparkContext(conf)
  //拿到sqlcontext对象，为了转换能parque文件
  val context: SQLContext = new SQLContext(sc)
  //读取文件
  private val parquet: DataFrame = context.read.parquet(load.getString("DataForParquet"))
  //处理数据
  private val map: RDD[(String, String, Int, Int, List[Double])] = parquet.map(t => {
    //第一个字段，运营商名字
    val ispname = t.getAs[String]("ispname")
    //第二个字段，网络类型
    val netname = t.getAs[String]("networkmannername")
    //第三个地段，设备类型
    val devicetype = t.getAs[Int]("devicetype")
    //第四个字段，操作系统
    val client = t.getAs[Int]("client")
    var requestmode = t.getAs[Int]("requestmode")
    var processnode = t.getAs[Int]("processnode")
    var iseffective = t.getAs[Int]("iseffective")
    var isbilling = t.getAs[Int]("isbilling")
    var isbid = t.getAs[Int]("isbid")
    var iswin = t.getAs[Int]("iswin")
    var adorderid = t.getAs[Int]("adorderid")
    //根据计算逻辑表格，拿出价格，下面使用
    val winprice = t.getAs[Double]("winprice")
    val adpayment = t.getAs[Double]("adpayment")
    //定义一个list用于聚合
    //list（原始请求,有效请求,广告请求,参与竞价数,竞价成功数,展示量,点击量,广告成本,广告消费）
    var list = ListBuffer[Double]()
    //开始判断
    if (requestmode == 1 && processnode >= 1) list.append(1) else list.append(0)
    if (requestmode == 1 && processnode >= 2) list.append(1) else list.append(0)
    if (requestmode == 1 && processnode == 3) list.append(1) else list.append(0)
    if (iseffective == 1 && isbilling == 1 && isbid == 1 && adorderid != 1) list.append(1) else list.append(0)
    if (iseffective == 1 && isbilling == 1 && iswin == 1) list.append(1) else list.append(0)
    if (requestmode == 2 && iseffective == 1) list.append(1) else list.append(0)
    if (requestmode == 3 && iseffective == 1) list.append(1) else list.append(0)
    if (iseffective == 1 && isbilling == 1 && iswin == 1) list.append(winprice / 1000) else list.append(0)
    if (iseffective == 1 && isbilling == 1 && iswin == 1) list.append(adpayment / 1000) else list.append(0)
    (ispname, netname, devicetype, client, list.toList)
  }).cache()
  //运营商聚合
  private val key = map.map(t => (t._1, t._5)).reduceByKey({
    (list1, list2) => list1.zip(list2).map(t => t._1 + t._2)
  })
  //网络类型聚合
  private val key1 = map.map(t => (t._2, t._5)).reduceByKey({
    (list1, list2) => list1.zip(list2).map(t => t._1 + t._2)
  })
  //设备类型聚合
  private val key2 = map.map(t => (t._3, t._5)).reduceByKey({
    (list1, list2) => list1.zip(list2).map(t => t._1 + t._2)
  })
  //操作系统聚合
  private val key3 = map.map(t => (t._4, t._5)).reduceByKey({
    (list1, list2) => list1.zip(list2).map(t => t._1 + t._2)
  })
  private val map1 = key2.map(t => {
    val s = if (t._1 == 1) "手机" else "平板"
    Row(s, t._2(0), t._2(1), t._2(2), t._2(3), t._2(4), t._2(5), t._2(6), t._2(7), t._2(8))
  })
  var schema =
    StructType(
      List(
        StructField("操作系统", StringType),
        StructField("原始请求", DoubleType),
        StructField("有效请求", DoubleType),
        StructField("广告请求", DoubleType),
        StructField("参与竞价数", DoubleType),
        StructField("竞价成功数", DoubleType),
        StructField("展示量", DoubleType),
        StructField("点击量", DoubleType),
        StructField("广告成本", DoubleType),
        StructField("广告消费", DoubleType)
      ))

  private val df = context.createDataFrame(map1, schema)
  df.show()

}
